package com.rickie.pvuv.message;

import com.alibaba.fastjson.JSONObject;
import com.rickie.pvuv.config.Constants;
import com.rickie.pvuv.domain.UserBehaviorEvent;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Calendar;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

public class KafkaMessageProducer {
    public static void main(String[] args) {
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", Constants.KAFKA_BOOTSTRAP_SERVER);
        kafkaParams.put("key.serializer", StringSerializer.class.getName());
        kafkaParams.put("value.serializer", StringSerializer.class.getName());
        String topic = Constants.TOPIC;

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(kafkaParams);
        int index = 0;
        while(index < 1000) {
            String msg = generateMessage();
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, msg);
            kafkaProducer.send(record);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            index++;
        }
        kafkaProducer.close();
    }

    /**
     * 构造消息
     */
    public static String generateMessage() {
        Calendar dateNow = Calendar.getInstance();
        dateNow.add(Calendar.DATE, new Random().nextInt(10));
        String[] action = {"homepage", "product", "cart", "login", "logout"};
        String[] category = {"c1", "c2", "c3", "c4", "c5"};
        UserBehaviorEvent userBehaviorEvent = new UserBehaviorEvent();
        userBehaviorEvent.setUserId(new Random().nextInt(10));
        userBehaviorEvent.setItemId(new Random().nextInt(10000));
        userBehaviorEvent.setCategory(category[new Random().nextInt(category.length)]);
        userBehaviorEvent.setAction(action[new Random().nextInt(action.length)]);
        //userBehaviorEvent.setTs(System.currentTimeMillis()/1000);
        userBehaviorEvent.setTs(dateNow.getTimeInMillis()/1000);
        String str = JSONObject.toJSONString(userBehaviorEvent);
        System.out.println(str);
        return str;
    }
}